package com.butterfly.hotkey.worker.framework.netty.pusher;


import cn.hutool.core.collection.CollectionUtil;
import com.butterfly.hotkey.core.constant.Constants;
import com.butterfly.hotkey.core.constant.MsgTypeConst;
import com.butterfly.hotkey.core.constant.NumConst;
import com.butterfly.hotkey.core.model.AppInfoModel;
import com.butterfly.hotkey.core.model.HotKeyModel;
import com.butterfly.hotkey.core.model.msg.HotKeyMsg;
import com.butterfly.hotkey.core.util.NettyMsgUtil;
import com.butterfly.hotkey.common.factory.CommonFactoryBean;
import com.butterfly.hotkey.worker.framework.component.WorkerContainer;
import com.google.common.collect.ImmutableListMultimap;
import com.google.common.collect.Multimaps;
import com.google.common.collect.Queues;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import static com.butterfly.hotkey.worker.constants.FactoryConstant.PUSHER;

/**
 * 推送给App端
 *
 * @author butterfly
 * @version 1.0
 */
@Component
@CommonFactoryBean(type = PUSHER, valueStr = "client")
public class ClientPusher implements IPusher {

    private final static Logger log = LoggerFactory.getLogger(Constants.WORKER);



    /**
     * 每次推送多少个hotkey
     */
    private static final int BATCH_SIZE = NumConst.ONE_HUNDRED;

    /**
     * QUEUE的超时时间
     */
    private static final int QUEUE_TIMEOUT = NumConst.TEN;

    /**
     * 热key队列
     */
    private static final LinkedBlockingQueue<HotKeyModel> QUEUE = new LinkedBlockingQueue<>();

    @Resource
    private TaskExecutor taskExecutor;

    @Resource
    private WorkerContainer workerContainer;

    @PostConstruct
    public void init() {
        taskExecutor.execute(() -> {
            while (true) {
                push();
            }
        });
    }

    @Override
    public void push(HotKeyModel model) {
        QUEUE.offer(model);
    }

    private void push() {
        try {
            List<HotKeyModel> waitPushModels = new ArrayList<>();
            // 当数据达到100条，或者10ms时，都进行推送
            Queues.drain(QUEUE, waitPushModels, BATCH_SIZE, QUEUE_TIMEOUT, TimeUnit.MILLISECONDS);
            if (CollectionUtil.isEmpty(waitPushModels)) {
                return;
            }

            // 将推送数据按照app进行分组
            ImmutableListMultimap<String, HotKeyModel> keyMap = Multimaps.index(waitPushModels, HotKeyModel::getAppName);
            for (Map.Entry<String, Collection<HotKeyModel>> entry : keyMap.asMap().entrySet()) {
                String appName = entry.getKey();
                Collection<HotKeyModel> hotkeys = entry.getValue();

                Map<String, AppInfoModel> clientMap = workerContainer.getClientMap();
                AppInfoModel appInfoModel = clientMap.get(appName);
                if (appInfoModel != null) {
                    HotKeyMsg hotKeyMsg = new HotKeyMsg(MsgTypeConst.RESPONSE_NEW_HOT_KEY);
                    hotKeyMsg.setHotKeyModels(hotkeys);
                    appInfoModel.writeAndFlush(NettyMsgUtil.buildByteBuf(hotKeyMsg));
                }
            }
        } catch (Exception e) {
            log.error("push hot key to app failed", e);
        }
    }

}
